package com.xian.java.stream;

//引用依赖模块的bean
import com.xian.bean.Person;
import com.xian.util.MyDateUtils;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

public class SplitSelectStream {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();

        List<Person> persons = new ArrayList<Person>();
        persons.add(new Person(1,"jemson", MyDateUtils.localDate2Date("2020-10-25")));
        persons.add(new Person(8,"杨幂", MyDateUtils.localDate2Date("2020-01-09")));
        persons.add(new Person(3,"张筱雨", MyDateUtils.localDate2Date("2020-11-23")));
        persons.add(new Person(5,"高圆圆", MyDateUtils.localDate2Date("2020-02-05")));

        DataSource<Person> personDataSource = env.fromCollection(persons);

        DataStreamSource<Person> personDataStreamSource = senv.fromCollection(persons);

        SplitStream<Person> splitStream = personDataStreamSource.split(new OutputSelector<Person>() {
            @Override
            public Iterable<String> select(Person person) {
                List<String> list = new ArrayList<String>(); //存标签
                if (person.getBirthdy().before(MyDateUtils.localDate2Date("2020-05-01"))) { //大于2020-05-01
                    list.add("gt");
                } else {
                    list.add("le"); //小于等于
                }
                return list;
            }
        });

        DataStream<Person> gtPersons = splitStream.select("gt");
        gtPersons.print("gt");  //打印在前面会有gt标识

        DataStream<Person> lePesons = splitStream.select("le");
        lePesons.print("le");

        splitStream.print("all");




        senv.execute();


    }
}
